package org.example.realtime.jtp.dws.log.job;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/**
 * @Author Lianzy
 * @Package org.example.realtime.jtp.dws.log.job
 * @Date 2025/5/21 14:37
 */
public class JtpTrafficAppStartMinuteWindowDwsJob {
    /**
     * {
     *   "common": {
     *     "ar": "440000",   //todo：ar ：可能是地区编码 ，代表用户所在地区，这里值为 440000 ，可能对应广东省。
     *     "uid": "663",   //todo：uid ：用户唯一标识 ，用于识别特定用户，值为 663 。
     *     "os": "Android 11.0",   //todo：os ：操作系统 ，这里是 Android 11.0 ，表明用户设备使用的操作系统版本。
     *     "ch": "huawei",   //todo：ch ：渠道 ，值为 huawei ，可能表示应用是从华为应用商店等华为相关渠道下载的。
     *     "is_new": "0",   //todo：is_new ：是否为新用户标识 ，0 一般表示不是新用户，1 可能代表新用户。
     *     "md": "vivo iqoo3",   //todo：md ：设备型号 ，vivo iqoo3 说明用户使用的设备是 vivo iqoo3 手机。
     *     "mid": "mid_671744",   //todo：mid ：可能是媒体识别码或中间标识 ，用于广告投放等场景下的设备或用户关联标识，值为 mid_671744 。
     *     "vc": "v2.1.134",   //todo：vc ：应用版本号 ，v2.1.134 表示应用当前版本。
     *     "ba": "vivo"   //todo：ba ：品牌 ，vivo 表明设备品牌是 vivo 。
     *   },
     *   "start": {
     *     "entry": "notice",       //todo：entry ：进入应用的入口 ，notice 表示是通过通知进入应用。
     *     "open_ad_skip_ms": 0,        //todo：open_ad_skip_ms ：跳过开屏广告的时间（毫秒） ，值为 0 说明没有跳过开屏广告操作，或没有开屏广告可跳过。
     *     "open_ad_ms": 7328,      //todo：open_ad_ms ：开屏广告展示时长（毫秒） ，值为 7328 ，即开屏广告展示了 7328 毫秒。
     *     "loading_time": 13428,       //todo：loading_time ：应用加载时长（毫秒） ，13428 表示从启动到加载完成花费了 13428 毫秒。
     *     "open_ad_id": 15     //todo：open_ad_id ：开屏广告 ID ，值为 15 ，用于标识具体的开屏广告内容。
     *   },
     *   "ts": 1746094055000  //todo：ts ：时间戳 。
     * }
     */
    public static void main(String[] args) {
        //执行环境-env
        TableEnvironment tEnv = getTableEnv();
        //输入表映射到kafka消息队列
        createInputTable(tEnv);
        //数据处理-select
        handle(tEnv);
        //输出表 映射到clickhouse
        createOutputTable(tEnv);
        //关联维表数据
        handleDim(tEnv);
        //保存数据到clickhouse
        saaveToClickhouse(tEnv);
    }

    private static void handleDim(TableEnvironment tabEnv) {

        Table startAdReportTable = tabEnv.sqlQuery("SELECT\n" +
                "    window_start AS window_start_time,\n" +
                "    window_end AS window_end_time,\n" +
                "    open_ad_id ,\n" +
                "    count(open_ad_id) AS open_ad_count ,\n" +
                "    sum(open_ad_ms) AS open_ad_ms ,\n" +
                "    sum(open_ad_skip_ms) AS open_ad_skip_ms,\n" +
                "    UNIX_TIMESTAMP() * 1000 AS ts ,\n" +
                "    PROCTIME() AS proc_time\n" +
                "FROM TABLE(\n" +
                "        TUMBLE(TABLE start_log_table, DESCRIPTOR(row_time), INTERVAL '10' second)\n" +
                "     )\n" +
                "GROUP BY window_start, window_end,open_ad_id"
        );

        tabEnv.createTemporaryView("start_ad_report_table", startAdReportTable);


        tabEnv.executeSql("CREATE TABLE IF NOT EXISTS ads_info_mysql_source_dim (\n" +
                "    id INT COMMENT '主键id',\n" +
                "    name String COMMENT'广告名称',\n" +
                "    content String COMMENT '广告内容（图片URL/视频URL/H5代码）',\n" +
                "    type String COMMENT '广告类型(IMAGE/VIDEO/HTML5)',\n" +
                "    start_time TIMESTAMP COMMENT '展示开始时间',\n" +
                "    end_time TIMESTAMP COMMENT'展示结束时间',\n" +
                "    is_active INT COMMENT'是否启用',\n" +
                "    target_condition String COMMENT '定向条件',\n" +
                "    created_at TIMESTAMP COMMENT '数据插入时间',\n" +
                "    updated_at TIMESTAMP COMMENT '数据更新时间',\n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                ")WITH (\n" +
                "    'connector' = 'jdbc',\n" +
                "    'url' = 'jdbc:mysql://node103:3306/jtp_ads',\n" +
                "    'table-name' = 'ads_info',\n" +
                "    'lookup.cache.max-rows' = '10',\n" +
                "    'lookup.cache.ttl' = '1 hour',\n" +
                "    'username' = 'root',\n" +
                "    'password' = '123456',\n" +
                "    'driver' = 'com.mysql.cj.jdbc.Driver'\n" +
                ")"
        );


        Table joinDimTable = tabEnv.sqlQuery("SELECT\n" +
                "    t1.window_start_time ,\n" +
                "    t1.window_end_time ,\n" +
                "    t2.name AS open_ad_name ,\n" +
                "    t1.open_ad_count ,\n" +
                "    t1.open_ad_ms ,\n" +
                "    t1.open_ad_skip_ms ,\n" +
                "    t1.ts\n" +
                "FROM start_ad_report_table t1\n" +
                "LEFT JOIN ads_info_mysql_source_dim FOR SYSTEM_TIME AS OF t1.proc_time AS t2\n" +
                "ON cast(t1.open_ad_id as int) = t2.id");


//        joinDimTable.execute().print();

        tabEnv.createTemporaryView("start_ad_report_table_r1", joinDimTable);
    }

    private static TableEnvironment getTableEnv() {
        //环境属性
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();
        TableEnvironment tEnv = TableEnvironment.create(settings);
        //配置属性
        Configuration configuration = tEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
//        configuration.setString("");
        // 返回对象
        return tEnv;
    }
    private static void createInputTable(TableEnvironment tEnv) {
        tEnv.executeSql("CREATE TABLE if not exists dwd_traffic_start_log_kafka_source\n" +
                "(\n   " +
                "`common` MAP<STRING, STRING> COMMENT '公共环境信息',\n" +
                "`start` MAP<STRING, STRING> COMMENT '启动信息',\n" +
                "`ts` BIGINT,\n" +
                "row_time AS TO_TIMESTAMP(FROM_UNIXTIME(ts / 1000, 'yyyy-MM-dd HH:mm:ss.SSS')),\n" +
                "WATERMARK FOR row_time AS row_time - INTERVAL '0' MINUTE\n" +
                ") WITH (\n" +
                "'connector' = 'kafka',\n" +
                "'topic' = 'dwd-log-startT-log',\n" +
                "'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                "'properties.group.id' = 'gid_dws_log_app_start',\n" +
                "'scan.startup.mode' = 'earliest-offset',\n" +
                "'format' = 'json',\n" +
                "'json.fail-on-missing-field' = 'false',\n" +
                "'json.ignore-parse-errors' = 'true'\n" +
                ")");
    }

    private static void handle(TableEnvironment tEnv) {
        //提取启动日志
        Table startLogTable = tEnv.sqlQuery("SELECT\n" +
                " `start`['entry'] AS entry\n" +
                ", CAST(`start`['loading_time'] AS BIGINT) AS loading_time\n" +
                ", `start`['open_ad_id'] AS open_ad_id\n" +
                ", CAST(`start`['open_ad_ms'] AS BIGINT) AS open_ad_ms\n" +
                ", CAST(`start`['open_ad_skip_ms'] AS BIGINT) AS open_ad_skip_ms\n" +
                ", row_time\n" +
                " FROM dwd_traffic_start_log_kafka_source\n" +
                " WHERE `start`['entry'] IS NOT NULL");
//        startLogTable.execute().print();
        tEnv.createTemporaryView("start_log_table", startLogTable);

        //设置窗口 分组聚合
        Table startReportTable = tEnv.sqlQuery("SELECT     " +
                "TUMBLE_START(row_time, INTERVAL '10' second) AS window_start_time     " +
                ", TUMBLE_END(row_time, INTERVAL '10' second) AS window_end_time     " +
                ", entry     " +
                ", count(entry) AS entry_count     " +
                ", sum(loading_time) AS loading_time     " +
                ", UNIX_TIMESTAMP() * 1000 AS ts FROM start_log_table GROUP BY     " +
                "TUMBLE(row_time, INTERVAL '10' second ),     entry"
        );
//        startReportTable.execute().print();
        tEnv.createTemporaryView("start_report_table", startReportTable);

//        Table startAdReportTable = tEnv.sqlQuery("SELECT     " +
//                "TUMBLE_START(row_time, INTERVAL '10' second) AS window_start_time      " +
//                ", TUMBLE_END(row_time, INTERVAL '10' second) AS window_end_time      " +
//                ", open_ad_id      " +
//                ", count(open_ad_id) AS open_ad_count      " +
//                ", sum(open_ad_ms) AS open_ad_ms      " +
//                ", sum(open_ad_skip_ms) AS open_ad_skip_ms      " +
//                ", UNIX_TIMESTAMP() * 1000 AS ts FROM start_log_table GROUP BY     " +
//                "TUMBLE(row_time, INTERVAL '10' second ),     open_ad_id");
//
////        startAdReportTable.execute().print();
//        tEnv.createTemporaryView("start_ad_report_table", startAdReportTable);

    }

    private static void saaveToClickhouse(TableEnvironment tEnv) {
        //        tabEnv.createTemporaryView("start_report_table_click",startReportTable);
        //保存app启动类型汇总表
        tEnv.executeSql("INSERT INTO dws_traffic_app_start_window_report_clickhouse_sink SELECT     " +
                "DATE_FORMAT(window_start_time, 'yyyy-MM-dd HH:mm:ss') AS window_start_time     " +
                ", DATE_FORMAT(window_end_time, 'yyyy-MM-dd HH:mm:ss') AS window_end_time     " +
                ", entry     " +
                ", entry_count     " +
                ", loading_time     " +
                ", ts FROM start_report_table"
        );

//        tabEnv.createTemporaryView("start_ad_report_table_click", startReportAdTable);
        //保存app启动广告汇总表
        tEnv.executeSql("INSERT INTO dws_traffic_app_start_ad_window_report_clickhouse_sink SELECT     " +
                "DATE_FORMAT(window_start_time, 'yyyy-MM-dd HH:mm:ss') AS window_start_time      " +
                ", DATE_FORMAT(window_end_time, 'yyyy-MM-dd HH:mm:ss') AS window_end_time      " +
                ", open_ad_name      " +
                ", open_ad_count      " +
                ", open_ad_ms      " +
                ", open_ad_skip_ms      " +
                ", ts FROM start_ad_report_table_r1"
        );
    }

    private static void createOutputTable(TableEnvironment tEnv) {
        //app启动类型汇总表
        tEnv.executeSql("CREATE TABLE dws_traffic_app_start_window_report_clickhouse_sink (   " +
                "`window_start_time` STRING COMMENT '窗口开始日期时间',   " +
                "`window_end_time` STRING COMMENT '窗口结束日期时间',   " +
                "`entry` STRING COMMENT '启动app方式',   " +
                "`entry_count` BIGINT COMMENT '启动app方式次数',   " +
                "`loading_time` BIGINT COMMENT '启动app方式次数',   " +
                "`ts` BIGINT COMMENT '启动app加载时间' " +
                ") WITH (   " +
                "'connector' = 'clickhouse',   " +
                "'url' = 'jdbc:clickhouse://node103:8123/jtp_log_report',   " +
                "'table' = 'dws_traffic_app_start_window_report',   " +
                "'username' = 'default',   " +
                "'password' = '',   " +
                "'format' = 'json' )"
        );

        //app启动广告汇总表
        tEnv.executeSql("CREATE TABLE dws_traffic_app_start_ad_window_report_clickhouse_sink (   " +
                "`window_start_time` STRING COMMENT '窗口开始日期时间',   " +
                "`window_end_time` STRING COMMENT '窗口结束日期时间',   " +
                "`open_ad_name` STRING COMMENT '启动app方式',   " +
                "`open_ad_count` BIGINT COMMENT '广告展示次数',   " +
                "`open_ad_ms` BIGINT COMMENT '广告总共播放时间',   " +
                "`open_ad_skip_ms` BIGINT COMMENT '用户跳过广告时间',   " +
                "`ts` BIGINT COMMENT '数据计算时间戳' " +
                ") WITH (   " +
                "'connector' = 'clickhouse',   " +
                "'url' = 'jdbc:clickhouse://node103:8123/jtp_log_report',   " +
                "'table' = 'dws_traffic_app_start_ad_window_report',   " +
                "'username' = 'default',   " +
                "'password' = '',   " +
                "'format' = 'json' )"
        );
    }


}
